package com.wang.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.wang.enums.DbEventTypeEnum;
import com.wang.service.ICanalService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;

@Service
@Slf4j
public class CanalServiceImpl implements ICanalService {

    @Autowired
    private CanalConnector connector;


    @Override
    public Message getMessageData(int batchSize) {
        Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
        return message;
    }

    @Override
    public List<JSONObject> getChangeDataList(int batchSize) {
        List<JSONObject> jsonObjectList = new ArrayList<>();
        Message message = this.getMessageData(batchSize);
        long batchId = message.getId();
        try {
            int size = message.getEntries().size();
            if (batchId != -1 || size != 0) {
                List<CanalEntry.Entry> entrys = message.getEntries();
                for (CanalEntry.Entry entry : entrys) {
                    if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                        continue;
                    }
                    CanalEntry.RowChange rowChage = null;
                    try {
                        rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                    } catch (Exception e) {
                        throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                                e);
                    }
                    CanalEntry.EventType eventType = rowChage.getEventType();
                    log.info(String.format("canal事件描述== binlog[%s:%s] , name[%s,%s] , eventType : %s",
                            entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                            entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                            eventType));

                    for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                        if (eventType == CanalEntry.EventType.DELETE) {
                            JSONObject deleteObj = getObjList(rowData.getBeforeColumnsList(), DbEventTypeEnum.DELETE);
                            jsonObjectList.add(deleteObj);
                        } else if (eventType == CanalEntry.EventType.INSERT) {
                            JSONObject insertObj = getObjList(rowData.getAfterColumnsList(), DbEventTypeEnum.INSERT);
                            jsonObjectList.add(insertObj);
                        } else {
                            //变更前
//                        jsonObjectList=   getObjList(rowData.getBeforeColumnsList());
                            //变更后
                            JSONObject updateObj = getObjList(rowData.getAfterColumnsList(), DbEventTypeEnum.UPDATE);
                            jsonObjectList.add(updateObj);
                        }
                    }
                }
            }
        } catch (Exception e) {
            log.error("解析出错了batchId=" + batchId);
        } finally {
            connector.ack(batchId);
        }

        return jsonObjectList;
    }

    public JSONObject getObjList(List<CanalEntry.Column> columns, DbEventTypeEnum dbEventType) {
        JSONObject jsonObject = new JSONObject();
        for (CanalEntry.Column column : columns) {
            log.info("canal数据变更字段：" + column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
            jsonObject.put(column.getName(), column.getValue());
        }
        jsonObject.put("dbEventType", dbEventType.getCode());
        return jsonObject;
    }

    @Override
    public void testData(int batchSize) {
        while (true){
            List<JSONObject> dataList = this.getChangeDataList(batchSize);
            dataList.forEach(d->{
              log.info("数据库变更对象："+d.toJSONString());
            });
        }

    }
}
